package com.kingsoft.dc.khaos.plugin.core.utils;

import org.apache.log4j.Logger;

import java.io.File;
import java.io.FileNotFoundException;
import java.io.FilenameFilter;

/**
 * Created by jing on 19/6/21.
 */
public class JarUtils {

    /**
     * <pre>
     * If there's a * specification in the "jar" argument (e.g. jar=./lib/*,./lib2/*),
     * this method helps to resolve the * into actual jar names inside the folder, and in order.
     * This is due to the requirement that Spark 1.4 doesn't seem to do the resolution for users
     *
     * </pre>
     *
     * @param unresolvedJarSpec
     * @return jar file list, comma separated, all .../* expanded into actual jar names in order
     */
    public static String resolveWildCardForJarSpec(String workingDirectory, String unresolvedJarSpec,
                                                   Logger log) {

        if (unresolvedJarSpec == null || unresolvedJarSpec.isEmpty())
            return "";

        StringBuilder resolvedJarSpec = new StringBuilder();

        String[] unresolvedJarSpecList = unresolvedJarSpec.split(",");
        for (String s : unresolvedJarSpecList) {
            // if need resolution
            if (s.endsWith("*")) {
                // remove last 2 characters to get to the folder
                String dirName = String.format("%s/%s", workingDirectory, s.substring(0, s.length() - 2));

                File[] jars = null;
                try {
                    jars = getFilesInFolderByRegex(new File(dirName), ".*jar");
                } catch (FileNotFoundException fnfe) {
                    log.warn("folder does not exist: " + dirName);
                    continue;
                }

                // if the folder is there, add them to the jar list
                for (File jar : jars) {
                    resolvedJarSpec.append(jar.toString()).append(",");
                }
            } else { // no need for resolution
                resolvedJarSpec.append(s).append(",");
            }
        }

        log.debug("resolveWildCardForJarSpec: resolvedJarSpec: " + resolvedJarSpec);

        // remove the trailing comma
        int lastCharIndex = resolvedJarSpec.length() - 1;
        if (lastCharIndex >= 0 && resolvedJarSpec.charAt(lastCharIndex) == ',') {
            resolvedJarSpec.deleteCharAt(lastCharIndex);
        }

        return resolvedJarSpec.toString();
    }

    /**
     * <pre>
     * This method looks for the proper user execution jar or a python file.
     * The user input is expected in the following 3 formats:
     *   1. ./lib/abc
     *   2. ./lib/abc.jar
     *   3. ./lib/abc.py
     *
     * This method will use prefix matching to find any jar/py that is the form of abc*.(jar|py),
     * so that users can bump jar versions without doing modifications to their Hadoop DSL.
     *
     * This method will throw an Exception if more than one jar that matches the prefix is found
     *
     * @param workingDirectory
     * @param userSpecifiedJarName
     * @return the resolved actual jar/py file name to execute
     */
    public static String resolveExecutionJarName(String workingDirectory,
                                                 String userSpecifiedJarName, Logger log) {

        if (log.isDebugEnabled()) {
            String debugMsg = String.format(
                    "Resolving execution jar name: working directory: %s,  user specified name: %s",
                    workingDirectory, userSpecifiedJarName);
            log.debug(debugMsg);
        }

        // in case user decides to specify with abc.jar, instead of only abc
        if (userSpecifiedJarName.endsWith(".jar")) {
            userSpecifiedJarName = userSpecifiedJarName.replace(".jar", "");
        } else if (userSpecifiedJarName.endsWith(".py")) {
            userSpecifiedJarName = userSpecifiedJarName.replace(".py", "");
        }

        // can't use java 1.7 stuff, reverting to a slightly ugly implementation
        String userSpecifiedJarPath = String.format("%s/%s", workingDirectory, userSpecifiedJarName);
        int lastIndexOfSlash = userSpecifiedJarPath.lastIndexOf("/");
        final String jarPrefix = userSpecifiedJarPath.substring(lastIndexOfSlash + 1);
        final String dirName = userSpecifiedJarPath.substring(0, lastIndexOfSlash);

        if (log.isDebugEnabled()) {
            String debugMsg = String.format("Resolving execution jar name: dirname: %s, jar name: %s",
                    dirName, jarPrefix);
            log.debug(debugMsg);
        }

        File[] potentialExecutionJarList;
        try {
            potentialExecutionJarList = getFilesInFolderByRegex(new File(dirName), jarPrefix + ".*(jar|py)");
        } catch (FileNotFoundException e) {
            throw new IllegalStateException(
                    "execution jar is suppose to be in this folder, but the folder doesn't exist: "
                            + dirName);
        }

        if (potentialExecutionJarList.length == 0) {
            throw new IllegalStateException("unable to find execution jar for Spark at path: "
                    + userSpecifiedJarPath + "*.(jar|py)");
        } else if (potentialExecutionJarList.length > 1) {
            throw new IllegalStateException(
                    "I find more than one matching instance of the execution jar at the path, don't know which one to use: "
                            + userSpecifiedJarPath + "*.(jar|py)");
        }

        String resolvedJarName = potentialExecutionJarList[0].toString();
        log.info("Resolving execution jar/py name: resolvedJarName: " + resolvedJarName);
        return resolvedJarName;
    }

    /**
     * @return a list of files in the given folder that matches the regex. It may be empty, but will
     * never return a null
     * @throws FileNotFoundException
     */
    private static File[] getFilesInFolderByRegex(File folder, final String regex)
            throws FileNotFoundException {
        // sanity check

        if (!folder.exists()) {
            throw new FileNotFoundException();

        }
        if (!folder.isDirectory()) {
            throw new IllegalStateException(
                    "execution jar is suppose to be in this folder, but the object present is not a directory: "
                            + folder);
        }

        File[] matchingFiles = folder.listFiles(new FilenameFilter() {
            @Override
            public boolean accept(File dir, String name) {
                if (name.matches(regex))
                    return true;
                else
                    return false;
            }
        });

        if (matchingFiles == null) {
            throw new IllegalStateException(
                    "the File[] matchingFiles variable is null.  This means an IOException occured while doing listFiles.  Please check disk availability and retry again");
        }

        return matchingFiles;
    }


    public static String resolvedJarWithAbsolutePathSpec(String baseDir, String unresolvedJarSpec,
                                                         Logger log) {

        if (unresolvedJarSpec == null || unresolvedJarSpec.isEmpty())
            return "";
        StringBuilder resolvedJarSpec = new StringBuilder();

        String[] unresolvedJarSpecList = unresolvedJarSpec.split(",");
        for (String s : unresolvedJarSpecList) {
            if (!s.startsWith(File.separator)) {
                s = baseDir + File.separator + s;
            }
            resolvedJarSpec.append(s).append(",");
        }
        // remove the trailing comma
        int lastCharIndex = resolvedJarSpec.length() - 1;
        if (lastCharIndex >= 0 && resolvedJarSpec.charAt(lastCharIndex) == ',') {
            resolvedJarSpec.deleteCharAt(lastCharIndex);
        }

        return resolvedJarSpec.toString();
    }



}
